package com.xhwl.data.event;

import com.alibaba.fastjson.JSONObject;
import com.xhwl.starter.config.IConfig;
import com.xhwl.common.pojo.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @Author: luyi
 * @Description: 空间更新事件通知
 * 在参数中台配置 CONFIG_CODE 并设置值为推送kafka的topic值即可
 * @Date: Create in 14:15 2021/6/17
 */
@Component
public class SpaceUpdateListener implements ApplicationListener<SpaceUpdateEvent> {

    private static final Logger log = LoggerFactory.getLogger(SpaceUpdateListener.class);

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private IConfig config;

    private static final String CONFIG_CODE = "SPACE_MESSAGE_UPDATE";

    @Async
    @Override
    public void onApplicationEvent(SpaceUpdateEvent spaceUpdateEvent) {

        Config projectConfig = config.getProject(CONFIG_CODE, spaceUpdateEvent.getProjectId());
        if(null == projectConfig){
            return ;
        }
        String topic = projectConfig.getValue().toString();
        log.info("收到消息 {} 准备推送至kafka {}", JSONObject.toJSONString(spaceUpdateEvent.getSpace()), topic);
        String massage = JSONObject.toJSONString(spaceUpdateEvent.getSpace());
        ListenableFuture<SendResult<String, Object>> inputMessage = kafkaTemplate.send(topic, massage);
        inputMessage.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("主题 = {}， 空间更新推送成功 = {}",topic, massage);
            }
            @Override
            public void onFailure(Throwable ex) {
                log.info("主题 = {}， 空间更新推送失败 = {}",topic, massage);
            }
        });
    }

}
